package com.wunong.smart.bigdata.flink.aggregate;

import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * @author created by zealot.zt
 */
public class CountAggregateFunction implements AggregateFunction<Integer, Integer, Integer> {

    private static final long serialVersionUID = 8272255605039385457L;

    /**
     * 创建一个累加器
     *
     * @return
     */
    @Override
    public Integer createAccumulator() {
        return 0;
    }

    /**
     * 累计器合并数据结果
     * 1. 单线程执行数据的合并
     * 2. 合并之后返回一个累计器结果
     *
     * @param value
     * @param accumulator
     * @return
     */
    @Override
    public Integer add(Integer value, Integer accumulator) {
        return value + accumulator;
    }

    /**
     * 从累加器获取迭代结果
     *
     * @param accumulator
     * @return
     */
    @Override
    public Integer getResult(Integer accumulator) {
        return accumulator;
    }

    /**
     * 合并多个累加器结果，多线程
     * 1. 适用于多线程的累计器数据合并
     * 2. 多个累加器合并最终为一个累加器
     *
     * @param a
     * @param b
     * @return
     */
    @Override
    public Integer merge(Integer a, Integer b) {
        return a + b;
    }

}